package drds.binlog.sink.group;

import drds.binlog.common.Authors;
import drds.binlog.data_object.EntryType;
import drds.binlog.sink.SinkException;
import drds.binlog.store.Event;
import drds.common.Author;
import drds.common.Else;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 相比于{@linkplain TimelineEventMergeGroupBarrier}，增加了按事务支持，会按照事务进行分库合并处理
 */
// 支持事务保留
@Author(name = Authors.LI_YANG)
public class TransactionTimelineEventMergeGroupBarrier extends TimelineEventMergeGroupBarrier
{

    //
    public static final int unsettled = 0;
    public static final int transaction = 1;
    public static final int not_transaction = 2;
    private ThreadLocal<Boolean> inTransaction = new ThreadLocal()
    {
        protected Object initialValue()
        {
            return false;
        }
    };
    /**
     * <pre>
     * 几种状态：
     * 0：初始状态，允许大家竞争
     * 1: 事务数据处理中
     * 2: 非事务数据处理中
     * </pre>
     */
    private AtomicInteger transactionState = new AtomicInteger(0);


    public TransactionTimelineEventMergeGroupBarrier(int groupSize)
    {
        super(groupSize);
    }

    public void awaitWhenIsNotPermit(Event event) throws InterruptedException
    {
        try
        {
            super.awaitWhenIsNotPermit(event);
        } catch (InterruptedException e)
        {
            // 出现线程中断，可能是因为关闭或者主备切换
            // 主备切换对应的事务尾会未正常发送，需要强制设置为事务结束，允许其他队列通过
            reset();
            throw e;
        }
    }

    public void awaitWhenIsNotPermit(Event event, long timeout, TimeUnit timeUnit) throws InterruptedException, TimeoutException
    {
        try
        {
            super.awaitWhenIsNotPermit(event, timeout, timeUnit);
        } catch (InterruptedException e)
        {
            // 出现线程中断，可能是因为关闭或者主备切换
            // 主备切换对应的事务尾会未正常发送，需要强制设置为事务结束，允许其他队列通过
            reset();
            throw e;
        }
    }

    @Override
    protected boolean isPermit(Event event, long executeDateTimeLong)
    {
        if (transactionState.intValue() == transaction && inTransaction.get())
        {
            return true;//事务处理中
        }
        if (transactionState.intValue() == transaction && !inTransaction.get())
        {
            return false;//状态改变中,继续进行test,直到状态成功
        }
        //
        if (transactionState.intValue() == unsettled)
        {
            boolean permit = super.isPermit(event, executeDateTimeLong);
            if (!permit)
            {
                return false;
            } else
            {
                // 可能第一条送过来的数据不为Begin，需要做判断处理，如果非事务，允许直接通过，比如DDL语句
                if (event.getEntryType() == EntryType.transaction_begin)
                {
                    if (transactionState.compareAndSet(unsettled, transaction))
                    {
                        inTransaction.set(true);
                        return true;
                    } else
                    {
                        return false;//状态改变中,继续进行test,直到状态成功
                    }
                } else
                {
                    // 非事务中
                    if (transactionState.compareAndSet(unsettled, not_transaction))
                    {
                        // 当基于zk-cursor启动的时候，拿到的第一个Event是TransactionEnd
                        return true; // 允许通过DDL/DCL/TransactionEnd
                    } else
                    {
                        return false;//状态改变中,继续进行test,直到状态成功
                    }
                }
            }
        }
        //
        if (transactionState.intValue() == not_transaction)
        {
            return false;//存在其他状态,继续进行test,直到状态成功
        }
        throw new Else();
    }

    public void removeEventExecuteDateTimeLongFromQueue(Event event)
    {
        super.removeEventExecuteDateTimeLongFromQueue(event);

        // 应该先判断not_transaction，再判断是否是事务尾，因为事务尾也可以导致txState的状态为2
        // 如果先判断事务尾，那么not_transaction的状态可能永远没机会被修改了，系统出现死锁
        if (transactionState.intValue() == not_transaction)
        {// 非事务中
            boolean compareAndSet = transactionState.compareAndSet(not_transaction, unsettled);
            if (compareAndSet == false)
            {
                throw new SinkException("state is not correct in non-transaction");
            }
            return;
        }
        if (transactionState.intValue() == transaction)
        {
            // 事务中
            if (event.getEntryType() == EntryType.transaction_end)
            {
                inTransaction.set(false); // 事务结束并且已经成功写入store，清理标记，进入重新排队判断，允许新的事务进入
                boolean compareAndSet = transactionState.compareAndSet(transaction, unsettled);
                //存在时间上面的不同步
                if (compareAndSet == false)
                {
                    throw new SinkException("state is not correct in transaction");
                }
            } else
            {
                //...do...状态不会改变
            }
            return;
        }
        if (transactionState.intValue() == unsettled)
        {
            throw new SinkException("state is not correct in transaction");
            //return;
        }


    }

    public void interrupt()
    {
        super.interrupt();
        reset();
    }

    // 重新设置状态
    private void reset()
    {
        inTransaction.remove();
        transactionState.set(0);// 重新置位
    }

}
